﻿using System.Text;
using AutoTest.Common;
using AutoTest.Common.Extension;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace AutoTest.Framework.RabbitMQ
{

    /// <summary>
    /// RabbitMQ帮助类
    /// </summary>
    public class RabbitMQHelper
    {
        #region MQ对象
        /// <summary>
        /// MQ连接
        /// </summary>
        protected IConnection rabbitMQConnect = null!;
        /// <summary>
        /// MQ的Channel
        /// </summary>
        protected IModel rabbitMQChannel = null!;

        /// <summary>
        /// MQ的配置
        /// </summary>
        protected RabbitMQConfig _config = null!;

        /// <summary>
        /// 消息处理错误最大重试次数
        /// </summary>
        private const int MaxRetryCount = 5;

        #endregion

        #region 对象管理
        /// <summary>
        /// 锁对象
        /// </summary>
        private static readonly object _lockObj = new object();
        /// <summary>
        /// MQ连接的管理（管理的实际上就是RabbitMQHelper的对象）
        /// </summary>
        private static System.Collections.Concurrent.ConcurrentDictionary<string, RabbitMQHelper> _instanceDictionary = new System.Collections.Concurrent.ConcurrentDictionary<string, RabbitMQHelper>();
        #endregion

        #region 实例方法
        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="uname"></param>
        /// <param name="pwd"></param>
        /// <param name="hostname"></param>
        /// <param name="direcotory"></param>
        /// <param name="isntanceType"></param>
        private RabbitMQHelper(string uname, string pwd, string hostname, string direcotory, EnumRabbitMQInstanceType isntanceType = EnumRabbitMQInstanceType.Sender)
        {
            //string direcotory = ConfigHelper.GetConfigSetting("PayAssistMQDirectory");

            _config = new RabbitMQConfig()
            {
                Account = uname,
                Pwd = pwd,
                Directory = direcotory,
                Host = hostname,
                InstanceType = isntanceType
            };

            ConnectionFactory factory = new ConnectionFactory
            {
                UserName = uname,//用户名
                Password = pwd,//密码
                HostName = hostname,//rabbitmq ip
                VirtualHost = direcotory //"cjtpaywg"
            };
            //创建连接
            rabbitMQConnect = factory.CreateConnection();
            //创建通道
            rabbitMQChannel = rabbitMQConnect.CreateModel();
        }

        /// <summary>
        /// 实例对象
        /// </summary>
        /// <param name="uname">用户名</param>
        /// <param name="pwd">密码</param>
        /// <param name="hostname">MQ地址</param>
        /// <param name="direcotory">目录</param>
        /// <param name="instanceType">实例类型，发送者还是接收者</param>
        /// <returns></returns>
        public static RabbitMQHelper Instance(string uname, string pwd, string hostname, string direcotory, EnumRabbitMQInstanceType instanceType = EnumRabbitMQInstanceType.Sender)
        {

            string key = $"{uname}_{pwd}_{hostname}_{direcotory}_{instanceType}";
            RabbitMQHelper obj = null!;
            if (_instanceDictionary.ContainsKey(key))
            {
                obj = _instanceDictionary[key];
            }
            else
            {
                obj = new RabbitMQHelper(uname, pwd, hostname, direcotory, instanceType);
                _instanceDictionary.TryAdd(key, obj);
            }
            return obj;
        }
        /// <summary>
        /// 实例化对象
        /// </summary>
        /// <param name="config"></param>
        /// <returns></returns>
        public static RabbitMQHelper Instance(RabbitMQConfig config)
        {
            string key = $"{config.Account}_{config.Pwd}_{config.Host}_{config.Directory}_{config.InstanceType}";
            RabbitMQHelper obj = null!;

            if (_instanceDictionary.ContainsKey(key))
            {
                obj = _instanceDictionary[key];
            }
            else
            {
                obj = new RabbitMQHelper(config.Account, config.Pwd, config.Host, config.Directory, config.InstanceType);
                _instanceDictionary.TryAdd(key, obj);
            }

            return obj;
        }

        #endregion


        #region 发送消息方法

        /// <summary>
        /// 发送消息到消息队列
        /// </summary>
        /// <param name="quene">队列名</param>
        /// <param name="message">消息（UTF-8编码）</param>
        /// <param name="durable">是否持久化</param>
        /// <param name="autodelete">是否自动删除</param>
        public void SendMQMsg(string quene, string message, bool durable = true, bool autodelete = false)
        {
            rabbitMQChannel.QueueDeclare(quene, durable, false, autodelete, null);
            var msg = new MQMsg(message);
            var sendBytes = Encoding.UTF8.GetBytes(msg.ToJson());
            rabbitMQChannel.BasicPublish("", quene, null, sendBytes);
        }

        protected void SendMQMsg(string quene, MQMsg msg, bool durable = true, bool autodelete = false)
        {
            rabbitMQChannel.QueueDeclare(quene, durable, false, autodelete, null);
            var sendBytes = Encoding.UTF8.GetBytes(msg.ToJson());
            rabbitMQChannel.BasicPublish("", quene, null, sendBytes);
        }

        /// <summary>
        /// 广播消息（通过交换机Exchange方式）
        /// </summary>
        /// <param name="exchange">交换机</param>
        /// <param name="message">消息（UTF-8编码）</param>
        /// <param name="durable">是否持久化</param>
        /// <param name="autodelete">是否自动删除</param>
        public void BroadCaseMsg(string exchange, string message, bool durable = true, bool autodelete = false)
        {
            rabbitMQChannel.ExchangeDeclare(exchange, "fanout", durable, autodelete, null);
            //var sendBytes = Encoding.UTF8.GetBytes(msg);
            var msg = new MQMsg(message);
            var sendBytes = Encoding.UTF8.GetBytes(msg.ToJson());
            //发布消息
            rabbitMQChannel.BasicPublish(exchange, "", null, sendBytes);
        }

        protected void BroadCaseMsg(string exchange, MQMsg msg, bool durable = true, bool autodelete = false)
        {
            rabbitMQChannel.ExchangeDeclare(exchange, "fanout", durable, autodelete, null);
            var sendBytes = Encoding.UTF8.GetBytes(msg.ToJson());
            //发布消息
            rabbitMQChannel.BasicPublish(exchange, "", null, sendBytes);
        }

        #endregion

        #region 接收消息

        /// <summary>
        /// 接收消息队列的消息
        /// </summary>
        /// <param name="quene">队列</param>
        /// <param name="action">收到消息后执行的方法</param>
        /// <param name="durable">是否持久化</param>
        /// <param name="autodelete">是否自动删除</param>
        public void RecieveMsg(string quene, Action<string> action, bool durable = true, bool autodelete = false)
        {
            EventingBasicConsumer consumer = new EventingBasicConsumer(rabbitMQChannel);
            consumer.Received += (ch, ea) =>
            {
                var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                MQMsg msg = message.DeserializeJson<MQMsg>();
                try
                {
                    action(msg.Msg);
                }
                catch (Exception)
                {
                    var sender = Instance(this._config.Account, this._config.Pwd, _config.Host, _config.Directory, EnumRabbitMQInstanceType.Sender);
                    if (msg.RetryCount >= MaxRetryCount)
                    {
                        sender.SendMQMsg($"{quene}-error", message, durable, autodelete);
                    }
                    else
                    {
                        msg.RetryCount++;
                        sender.SendMQMsg(quene, msg, durable, autodelete);
                    }
                }

                rabbitMQChannel.BasicAck(ea.DeliveryTag, false);
            };
            rabbitMQChannel.QueueDeclare(quene, durable, false, autodelete, null);
            rabbitMQChannel.BasicQos(0, 1, false);
            rabbitMQChannel.BasicConsume(quene, false, consumer);
        }

        /// <summary>
        /// 接收广播消息
        /// </summary>
        /// <param name="exchange">交换机（Exchange）</param>
        /// <param name="action">收到消息后执行的方法</param>
        public void RecieveBroadCast(string exchange, Action<string> action)
        {
            rabbitMQChannel.ExchangeDeclare(exchange, "fanout", true, false);
            // binding
            EventingBasicConsumer consumer = new EventingBasicConsumer(rabbitMQChannel);
            string quene = $"{exchange}-{Guid.NewGuid().ToString("N")}";
            consumer.Received += (sender, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                MQMsg msg = message.DeserializeJson<MQMsg>();
                try
                {
                    if (action != null)
                    {
                        action(msg.Msg);
                    }
                }
                catch (Exception)
                {
                    var msgSender = Instance(this._config.Account, this._config.Pwd, _config.Host, _config.Directory, EnumRabbitMQInstanceType.Sender);
                    if (msg.RetryCount >= MaxRetryCount)
                    {
                        msgSender.SendMQMsg($"ex-{exchange}-error", message);
                    }
                    else
                    {
                        msg.RetryCount++;
                        msgSender.SendMQMsg(quene, msg);
                    }
                }
                rabbitMQChannel.BasicAck(ea.DeliveryTag, false);
            };
            rabbitMQChannel.QueueDeclare(quene, false, false, true, null);
            rabbitMQChannel.QueueBind(quene, exchange, "");
            rabbitMQChannel.BasicQos(0, 1, false);
            rabbitMQChannel.BasicConsume(quene, false, consumer);
        }

        /// <summary>
        /// 接受广播消息
        /// </summary>
        /// <param name="exchange">交换机（Exchange）</param>
        /// <param name="quene">队列</param>
        /// <param name="action">收到消息执行的方法</param>
        /// <param name="durable">队列是否持久化</param>
        /// <param name="autodelete">队列是否自动删除</param>
        public void RecieveBroadCast(string exchange, string quene, Action<string> action, bool durable = true, bool autodelete = false)
        {
            rabbitMQChannel.ExchangeDeclare(exchange, "fanout", true, false);
            // binding
            EventingBasicConsumer consumer = new EventingBasicConsumer(rabbitMQChannel);
            consumer.Received += (sender, ea) =>
            {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body.ToArray());
                MQMsg msg = message.DeserializeJson<MQMsg>();
                try
                {
                    if (action != null)
                    {
                        action(msg.Msg);
                    }
                }
                catch (Exception)
                {
                    var msgSender = Instance(this._config.Account, this._config.Pwd, _config.Host, _config.Directory, EnumRabbitMQInstanceType.Sender);
                    if (msg.RetryCount >= MaxRetryCount)
                    {
                        msgSender.SendMQMsg($"ex-{exchange}-error", message);
                    }
                    else
                    {
                        msg.RetryCount++;
                        msgSender.SendMQMsg(quene, msg, durable, autodelete);
                    }
                }
                rabbitMQChannel.BasicAck(ea.DeliveryTag, false);
            };
            rabbitMQChannel.QueueDeclare(quene, durable, false, autodelete, null);
            rabbitMQChannel.QueueBind(quene, exchange, "");
            rabbitMQChannel.BasicQos(0, 1, false);
            rabbitMQChannel.BasicConsume(quene, false, consumer);
        }


        #endregion

        /// <summary>
        /// 获取缓存的Key
        /// </summary>
        /// <returns></returns>
        private string GetCacheKey()
        {
            string key = $"{_config.Account}_{_config.Pwd}_{_config.Host}_{_config.Directory}_{_config.InstanceType}";
            return key;
        }

        /// <summary>
        /// 移除管理实例
        /// </summary>
        public RabbitMQHelper RemoveInstance()
        {
            RabbitMQHelper obj = null!;
            string key = GetCacheKey();
            if (key.IsNotEmpty() && _instanceDictionary.ContainsKey(key))
            {
                _instanceDictionary.TryRemove(key, out obj!);
            }
            return obj!;
        }

    }

}
